package belf.migrate.engine.job.execute;

import belf.migrate.api.job.Job;
import belf.migrate.api.table.schema.AbstractSinkCatalog;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.type.LogLevelType;
import belf.migrate.api.util.PostLog;
import belf.migrate.engine.BelfEvn;
import belf.migrate.engine.job.*;
import belf.migrate.engine.job.execute.helper.ExecuteSqlTool;
import belf.migrate.engine.job.model.execute.ExecuteResult;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;

import static belf.migrate.api.job.JobCode.*;

@Slf4j
public class ExecuteJobRunner extends BaseJobRunner {


    private static Gson gson = new Gson();

    public ExecuteJobRunner(Job job) {
        super(job);
    }


    @Override
    public JobResult check() {
        JobResult jobResult = new JobResult(SUCCESS, "任务执行成功");

        jobContext = new JobContext("" + job.getJobId(), taskConf);
        // 创建目标catalog
        AbstractSinkCatalog sinkCatalog = CatalogCreator.createSinkCatalog(taskConf);
        if (null == sinkCatalog) {
            log.error("No Sink Catalog for {}, Stop Job {}!",
                    taskConf.getSinkConf().getCatalogName(),
                    job.getJobId());

            jobResult = new VerifyJobResult(NO_SINK_CATALOG, "提交Job失败，没有找到源端" +
                    taskConf.getSourceConf().getCatalogName() +
                    "可适配的Catalog类，请联系开发者。");
            return jobResult;
        }
        jobContext.setSinkCatalog(sinkCatalog);
        BelfEvn.getInstance().addJobContext("" + job.getJobId(), jobContext);

        return jobResult;
    }

    @Override
    public JobResult execute() {
        PostLog.sendLogModel(
                job.getJobId(),
                job.getTask().getTaskType().getName(),
                "执行sql开始！",
                LogLevelType.INFO.getType(),
                EXECUTING,
                jobContext.getProgress());
        log.info("开始执行sql！");
        JobResult jobResult = check();
        if (jobResult.getCode() != SUCCESS) {
            return jobResult;
        }
        ExecuteResult result = new ExecuteSqlTool(jobContext, job).run();
        jobResult.setData(gson.toJson(result));
        jobResult.setJobId(job.getJobId());
        //释放数据库连接
        jobContext.release();
        log.info("执行sql结束！");
        PostLog.sendLogModel(
                job.getJobId(),
                job.getTask().getTaskType().getName(),
                "执行sql结束！",
                LogLevelType.INFO.getType(),
                EXECUTING,
                jobContext.getProgress());

        return jobResult;
    }
}
